
package com.leezai.mqtt.handler;

import com.leezai.mqtt.dto.MessageWrapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

public class MessageQueueHandler implements MessageHandler{

    private final BlockingQueue<MessageWrapper> messageQueue = new LinkedBlockingQueue<>(1000);
    private ExecutorService executorService;
    private static final Logger log = LoggerFactory.getLogger(MessageQueueHandler.class);


    //如果消费者一个线程不够，可以使用线程池
    //private final ExecutorService executorService = Executors.newFixedThreadPool(5);

    private final MessageProcessor processImpl;

    // 构造函数注入依赖
    public MessageQueueHandler(MessageProcessor processorImpl) {
        if (processorImpl == null){
            this.processImpl = new MessageProcessorImpl();
        }else{
            this.processImpl = processorImpl;
        }
    }


    /**
     * 提交消息到队列
     */
    public void addMessage(MessageWrapper message) {
        try {
            boolean success = messageQueue.offer(message);
            if (!success) {
                //塞数据库
                // 队列已满，无法放入新消息
                log.error("队列已满，消息被丢弃");
            }
            //不用put 阻塞就丢弃
//            messageQueue.put(message);
        } catch (Exception e) {

            log.error(e.getMessage());
        }
    }

    /**
     * 启动消息处理线程（延迟初始化）
     */
    public synchronized void startProcessing() {
        if (executorService == null || executorService.isShutdown()) {
            executorService = Executors.newSingleThreadExecutor();
            executorService.submit(this::processMessages);
            log.info("leezai.mqtt 消息处理线程已启动");
        }
    }


    /**
     * 消息处理逻辑
     */
    private void processMessages() {
        while (!Thread.currentThread().isInterrupted()) {
            try {
                MessageWrapper wrapper = messageQueue.take();
                String topic = wrapper.getTopic();
                byte[] message = wrapper.getMessage();
                processImpl.handleMessage(topic, message);
                // 处理消息逻辑
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                log.error("leezai.mqtt 消息处理线程被中断");
                break;
            } catch (Exception e) {
                log.error("leezai.mqtt队列消息处理异常:" + e.getMessage());
            }
        }
    }


    //关闭前销毁线程池
    public void shutdown() {
        executorService.shutdownNow();
        log.info("leezai.mqtt.MqttMessageQueueHandler 已关闭");
    }

    /**
     * 获取队列中待处理消息数量
     * @return 队列大小
     */
    public int getQueueSize() {
        return messageQueue.size();
    }

    /**
     * 检查消息处理器是否正在运行
     * @return true if processing, false otherwise
     */
    public boolean isProcessing() {
        return executorService != null && !executorService.isShutdown();
    }
}